package cn.juque.jdfsystem.config;

import cn.juque.common.base.DisruptorBody;
import cn.juque.common.base.DisruptorMessage;
import cn.juque.common.disruptor.DisruptorHandler;
import cn.juque.common.disruptor.MqEventFactory;
import cn.juque.common.disruptor.MqEventHandler;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author nuoka
 * @version 1.0.0
 * <li>IntelliJ IDEA</li>
 * <li></li>
 * @date 2021/8/24 23:05
 **/
@Configuration
@Slf4j
public class DisruptorConfig {

    @Bean
    public DisruptorHandler disruptorHandler() {
        return new DisruptorHandler();
    }

    @Bean("disruptorRingBuffer")
    public RingBuffer<DisruptorMessage<? extends DisruptorBody>> disruptorRingBuffer() {
        // 定义线程池
        DefaultThreadFactory threadFactory = new DefaultThreadFactory("disruptor-thread-pool");
        // 指定事件工厂
        MqEventFactory factory = new MqEventFactory();
        // 指定ringBuffer字节大小，必须是2的N次方
        int bufferSize = 1024 * 256;
        // 单线程模式，获取额外的性能
        Disruptor<DisruptorMessage<? extends DisruptorBody>> disruptor = new Disruptor<>(factory, bufferSize, threadFactory, ProducerType.SINGLE, new BlockingWaitStrategy());
        // 设置消费者
        disruptor.handleEventsWith(new MqEventHandler());
        // 启动disruptor线程
        disruptor.start();
        log.info("init disruptor successful");
        // 获取ringBuffer环，用于接收生产者
        return disruptor.getRingBuffer();
    }
}
